package com.atguigu.gmallrealtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmallrealtime.bean.TableProcess;
import com.atguigu.gmallrealtime.util.JDBCUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.checkerframework.checker.units.qual.A;

import java.sql.Connection;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Set;

/**
 * @author yhm
 * @create 2023-10-07 10:42
 */
public class DwdBroadcastFunction extends BroadcastProcessFunction<JSONObject, JSONObject, JSONObject> {

    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;
    private HashMap<String, TableProcess> hashMap = new HashMap<>();

    public DwdBroadcastFunction(MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        // 对配置表进行初始化加载  避免主流数据到的过快
        Connection mysqlConnection = JDBCUtil.getMysqlConnection();
        List<TableProcess> tableProcesses = JDBCUtil.queryList(mysqlConnection, "select * from gmall_config.table_process where sink_type='dwd'", TableProcess.class, true);
        for (TableProcess tableProcess : tableProcesses) {
            hashMap.put(tableProcess.getSourceTable() + tableProcess.getSourceType() , tableProcess);
        }

        mysqlConnection.close();
    }

    @Override
    public void processBroadcastElement(JSONObject value, Context ctx, Collector<JSONObject> out) throws Exception {
        // 将需要保留的事实表类型存放到状态中
        // 状态格式: key -> source_table+source_type  value-> tableProcess
        // 1.获取状态
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);

        // 2.读取配置表数据的op
        String op = value.getString("op");
        if ("d".equals(op)){
            TableProcess before = value.getObject("before", TableProcess.class);
            if ("dwd".equals(before.getSinkType())){
                String key = before.getSourceTable() + before.getSourceType();
                broadcastState.remove(key);
                hashMap.remove(key);
            }
        }else if ("u".equals(op)){
            // 删除之前的  创建之后的
            TableProcess before = value.getObject("before", TableProcess.class);
            if ("dwd".equals(before.getSinkType())){
                String key = before.getSourceTable() + before.getSourceType();
                broadcastState.remove(key);
                hashMap.remove(key);
            }
            TableProcess after = value.getObject("after", TableProcess.class);
            if("dwd".equals(after.getSinkType())){
                String afterKey = after.getSourceTable() + after.getSourceType();
                broadcastState.put(afterKey,after);
            }
        }else {
            TableProcess after = value.getObject("after", TableProcess.class);
            if("dwd".equals(after.getSinkType())){
                String afterKey = after.getSourceTable() + after.getSourceType();
                broadcastState.put(afterKey,after);
            }
        }
    }

    @Override
    public void processElement(JSONObject value, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        // 读取状态中的内容  对主流数据进行过滤 得到想要的事实表数据
        String tableName = value.getString("table");
        String type = value.getString("type");
        String key = tableName + type;
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        TableProcess tableProcess = broadcastState.get(key);
        if (tableProcess == null){
            // 可能是主流数据来的太早了  需要到初始化的map中确认
            tableProcess = hashMap.get(key);
        }

        if (tableProcess != null){
            // 当前为需要的数据
            JSONObject data = value.getJSONObject("data");
            String[] split = tableProcess.getSinkColumns().split(",");
            Set<String> dataKeys = data.keySet();
            dataKeys.removeIf(k -> !Arrays.asList(split).contains(k));
            data.put("sink_table",tableProcess.getSinkTable());
            out.collect(data);
        }
    }
}
